/**
 * Copyright (C) 2011-2012 Turn, Inc.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.turn.ttorrent.tracker;

import com.turn.ttorrent.Constants;
import com.turn.ttorrent.bcodec.BEValue;
import com.turn.ttorrent.common.LoggerUtils;
import com.turn.ttorrent.common.Peer;
import com.turn.ttorrent.common.TorrentLoggerFactory;
import com.turn.ttorrent.common.protocol.AnnounceRequestMessage;
import com.turn.ttorrent.common.protocol.TrackerMessage.ErrorMessage;
import com.turn.ttorrent.common.protocol.TrackerMessage.MessageValidationException;
import com.turn.ttorrent.common.protocol.http.HTTPAnnounceRequestMessage;
import com.turn.ttorrent.common.protocol.http.HTTPAnnounceResponseMessage;
import com.turn.ttorrent.common.protocol.http.HTTPTrackerErrorMessage;
import org.simpleframework.http.Status;
import org.slf4j.Logger;

import java.io.*;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import java.net.URL;
import java.net.HttpURLConnection;


/**
 * Tracker service to serve the tracker's announce requests.
 * <p>
 * <p>
 * It only serves announce requests on /announce, and only serves torrents the
 * {@link Tracker} it serves knows about.
 * </p>
 * <p>
 * <p>
 * The list of torrents {@see #requestHandler.getTorrentsMap()} is a map of torrent hashes to their
 * corresponding Torrent objects, and is maintained by the {@link Tracker} this
 * service is part of. The TrackerRequestProcessor only has a reference to this map, and
 * does not modify it.
 * </p>
 *
 * @author mpetazzoni
 * @see <a href="http://wiki.theory.org/BitTorrentSpecification">BitTorrent protocol specification</a>
 */
public class TrackerRequestProcessor {

    private static final Logger logger =
            TorrentLoggerFactory.getLogger(TrackerRequestProcessor.class);

    // 在这里定义边缘服务器地址常量
    private static final String FLASK_EDGE_SERVER = "http://127.0.0.1:5000";

    /**
     * The list of announce request URL fields that need to be interpreted as
     * numeric and thus converted as such in the request message parsing.
     */
    private static final String[] NUMERIC_REQUEST_FIELDS =
            new String[]{
                    "port", "uploaded", "downloaded", "left",
                    "compact", "no_peer_id", "numwant"
            };
    private static final int SEEDER_ANNOUNCE_INTERVAL = 150;

    private boolean myAcceptForeignTorrents = true; //default to true
    private int myAnnounceInterval = 60; //default value
    private final AddressChecker myAddressChecker;
    private final TorrentsRepository myTorrentsRepository;

    private String trackerName;

    // 添加getter和setter方法
    public String getTrackerName() {
        return trackerName;
    }

    public void setTrackerName(String trackerName) {
        this.trackerName = trackerName;
    }


    /**
     * Create a new TrackerRequestProcessor serving the given torrents.
     */
    public TrackerRequestProcessor(TorrentsRepository torrentsRepository) {
        this(torrentsRepository, new AddressChecker() {
            @Override
            public boolean isBadAddress(String ip) {
                return false;
            }
        });
    }

    public TrackerRequestProcessor(TorrentsRepository torrentsRepository, AddressChecker addressChecker) {
        myTorrentsRepository = torrentsRepository;
        myAddressChecker = addressChecker;
    }

    /**
     * Process the announce request.
     * <p>
     * <p>
     * This method attemps to read and parse the incoming announce request into
     * an announce request message, then creates the appropriate announce
     * response message and sends it back to the client.
     * </p>
     */
    public void process(final String uri, final String hostAddress, String passkey, RequestHandler requestHandler)
            throws IOException {
        // Prepare the response headers.

        /**
         * Parse the query parameters into an announce request message.
         *
         * We need to rely on our own query parsing function because
         * SimpleHTTP's Query map will contain UTF-8 decoded parameters, which
         * doesn't work well for the byte-encoded strings we expect.
         */
        HTTPAnnounceRequestMessage announceRequest;
        try {
            announceRequest = this.parseQuery(uri, hostAddress, passkey);
        } catch (MessageValidationException mve) {
            LoggerUtils.warnAndDebugDetails(logger, "Unable to parse request message. Request url is {}", uri, mve);
            serveError(Status.BAD_REQUEST, mve.getMessage(), requestHandler);
            return;
        }
        // 验证passkey
        if (announceRequest.getPasskey() == null || !validatePasskey(announceRequest.getPasskey())) {
            // 使用Ttorrent的HTTPTrackerErrorMessage
            ByteBuffer errorData = HTTPTrackerErrorMessage.craft("Invalid passkey or tracker inactive").getData();
            requestHandler.serveResponse(Status.OK.getCode(), "OK", errorData);
            return;
        }

        AnnounceRequestMessage.RequestEvent event = announceRequest.getEvent();

        if (event == null) {
            event = AnnounceRequestMessage.RequestEvent.NONE;
        }
        TrackedTorrent torrent = myTorrentsRepository.getTorrent(announceRequest.getHexInfoHash());

        // The requested torrent must be announced by the tracker if and only if myAcceptForeignTorrents is false
        if (!myAcceptForeignTorrents && torrent == null) {
            logger.warn("Requested torrent hash was: {}", announceRequest.getHexInfoHash());
            serveError(Status.BAD_REQUEST, ErrorMessage.FailureReason.UNKNOWN_TORRENT, requestHandler);
            return;
        }

        final boolean isSeeder = (event == AnnounceRequestMessage.RequestEvent.COMPLETED)
                || (announceRequest.getLeft() == 0);

        if (myAddressChecker.isBadAddress(announceRequest.getIp())) {
            if (torrent == null) {
                writeEmptyResponse(announceRequest, requestHandler);
            } else {
                writeAnnounceResponse(torrent, null, isSeeder, requestHandler);
            }
            return;
        }

        final Peer peer = new Peer(announceRequest.getIp(), announceRequest.getPort());

        System.out.println(String.format("%tT", new java.util.Date()) +
                " - Tracker: " + (this.trackerName != null ? this.trackerName : "unknown") +
                " - Passkey: " + (announceRequest.getPasskey() != null ? announceRequest.getPasskey() : "none") +
                " - Peer: " + announceRequest.getHexPeerId() +
                " - InfoHash: " + announceRequest.getHexInfoHash() +
                " - Uploaded: " + announceRequest.getUploaded() +
                " - Downloaded: " + announceRequest.getDownloaded() +
                " - Left: " + announceRequest.getLeft());

// 向Flask边缘服务器发送数据
        try {
            sendStatsToFlask(
                    announceRequest.getPasskey(),
                    announceRequest.getHexPeerId(),
                    announceRequest.getHexInfoHash(),
                    announceRequest.getUploaded(),
                    announceRequest.getDownloaded(),
                    announceRequest.getLeft()
            );
        } catch (Exception e) {
            // 记录错误但不中断正常流程
            logger.warn("Failed to send stats to Flask server: {}", e.getMessage());
        }


        try {
            torrent = myTorrentsRepository.putIfAbsentAndUpdate(announceRequest.getHexInfoHash(),
                    new TrackedTorrent(announceRequest.getInfoHash()),
                    event,
                    ByteBuffer.wrap(announceRequest.getPeerId()),
                    announceRequest.getHexPeerId(),
                    announceRequest.getIp(),
                    announceRequest.getPort(),
                    announceRequest.getUploaded(),
                    announceRequest.getDownloaded(),
                    announceRequest.getLeft());
        } catch (IllegalArgumentException iae) {
            LoggerUtils.warnAndDebugDetails(logger, "Unable to update peer torrent. Request url is {}", uri, iae);
            serveError(Status.BAD_REQUEST, ErrorMessage.FailureReason.INVALID_EVENT, requestHandler);
            return;
        }

        // Craft and output the answer
        writeAnnounceResponse(torrent, peer, isSeeder, requestHandler);
    }

    private void writeEmptyResponse(HTTPAnnounceRequestMessage announceRequest, RequestHandler requestHandler) throws IOException {
        HTTPAnnounceResponseMessage announceResponse;
        try {
            announceResponse = HTTPAnnounceResponseMessage.craft(
                    myAnnounceInterval,
                    0,
                    0,
                    Collections.<Peer>emptyList(),
                    announceRequest.getHexInfoHash());
            requestHandler.serveResponse(Status.OK.getCode(), Status.OK.getDescription(), announceResponse.getData());
        } catch (Exception e) {
            serveError(Status.INTERNAL_SERVER_ERROR, e.getMessage(), requestHandler);
        }
    }

    public void setAnnounceInterval(int announceInterval) {
        myAnnounceInterval = announceInterval;
    }

    public int getAnnounceInterval() {
        return myAnnounceInterval;
    }

    private boolean validatePasskey(String passkey) {
        // 如果没有传递passkey，则不允许访问
        if (passkey == null || passkey.isEmpty()) {
            logger.warn("尝试使用空passkey访问");
            return false;
        }

        try {
            // 创建URL连接到边缘服务器的验证接口，添加trackerName参数
            URL url = new URL(String.format("%s/api/passkey/validate/%s?tracker_name=%s",
                    FLASK_EDGE_SERVER,
                    passkey,
                    URLEncoder.encode(this.trackerName != null ? this.trackerName : "unknown-tracker", "UTF-8")));

            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("GET");
            conn.setRequestProperty("Accept", "application/json");
            conn.setConnectTimeout(3000); // 3秒连接超时
            conn.setReadTimeout(3000);    // 3秒读取超时

            // 获取响应
            int responseCode = conn.getResponseCode();

            if (responseCode == HttpURLConnection.HTTP_OK) {
                // 解析响应内容
                BufferedReader in = new BufferedReader(new InputStreamReader(conn.getInputStream()));
                String inputLine;
                StringBuilder response = new StringBuilder();

                while ((inputLine = in.readLine()) != null) {
                    response.append(inputLine);
                }
                in.close();

                // 解析JSON响应
                String jsonResponse = response.toString();
                // 检查是否包含"valid":true
                boolean isValid = jsonResponse.contains("\"valid\":true");

                // 记录验证结果，包括消息信息
                if (!isValid && jsonResponse.contains("\"message\":")) {
                    // 提取message字段的值进行日志记录
                    int messageStart = jsonResponse.indexOf("\"message\":\"") + 11;
                    int messageEnd = jsonResponse.indexOf("\"", messageStart);
                    if (messageStart >= 11 && messageEnd > messageStart) {
                        String errorMessage = jsonResponse.substring(messageStart, messageEnd);
                        logger.warn("Passkey验证失败: {} - {}", passkey, errorMessage);
                    }
                }

                logger.debug("Passkey验证结果: {} - {}", passkey, isValid ? "有效" : "无效");
                return isValid;
            } else {
                logger.warn("验证Passkey时收到非200响应: {}", responseCode);
                return false; // 如果服务器返回非200，视为验证失败
            }
        } catch (Exception e) {
            // 记录异常，但不中断正常流程
            logger.warn("验证Passkey时发生错误: {}", e.getMessage());
            if (logger.isDebugEnabled()) {
                logger.debug("详细错误:", e);
            }

            // 如果连接边缘服务器失败，我们选择拒绝请求
            return false;
        }
    }

    private void sendStatsToFlask(String passkey, String peerId, String infoHash,
                                  long uploaded, long downloaded, long left) {
        try {
            // 创建URL连接
            URL url = new URL(FLASK_EDGE_SERVER + "/api/stats/update"); // 使用常量
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("POST");
            conn.setRequestProperty("Content-Type", "application/json");
            conn.setRequestProperty("Accept", "application/json");
            conn.setDoOutput(true);

            // 创建JSON数据，添加trackerName字段
            String jsonInputString = String.format(
                    "{\"tracker\":\"%s\",\"passkey\":\"%s\",\"peerid\":\"%s\",\"infohash\":\"%s\",\"uploaded\":%d,\"downloaded\":%d,\"left\":%d}",
                    this.trackerName != null ? this.trackerName : "unknown-tracker",
                    passkey != null ? passkey : "",
                    peerId,
                    infoHash,
                    uploaded,
                    downloaded,
                    left
            );

            // 发送数据
            OutputStream os = null;
            try {
                os = conn.getOutputStream();
                byte[] input = jsonInputString.getBytes("utf-8");
                os.write(input, 0, input.length);
            } finally {
                if (os != null) {
                    try {
                        os.close();
                    } catch (IOException e) {
                        // 忽略关闭时的异常
                    }
                }
            }

            // 获取响应
            int responseCode = conn.getResponseCode();
            if (responseCode != HttpURLConnection.HTTP_OK) {
                logger.warn("Flask server returned non-OK status: {}", responseCode);
            }

            // 关闭连接
            conn.disconnect();
        } catch (Exception e) {
            // 记录错误日志，但不中断tracker的正常操作
            logger.warn("Error sending stats to Flask: {}", e.getMessage());
            if (logger.isDebugEnabled()) {
                // 替代LoggerUtils.debugDetails的方式
                logger.debug("Stack trace:", e);
            }
        }
    }

    private void writeAnnounceResponse(TrackedTorrent torrent, Peer peer, boolean isSeeder, RequestHandler requestHandler) throws IOException {
        HTTPAnnounceResponseMessage announceResponse;
        try {
            announceResponse = HTTPAnnounceResponseMessage.craft(
                    isSeeder ? SEEDER_ANNOUNCE_INTERVAL : myAnnounceInterval,
                    torrent.seeders(),
                    torrent.leechers(),
                    isSeeder ? Collections.<Peer>emptyList() : torrent.getSomePeers(peer),
                    torrent.getHexInfoHash());
            requestHandler.serveResponse(Status.OK.getCode(), Status.OK.getDescription(), announceResponse.getData());
        } catch (Exception e) {
            serveError(Status.INTERNAL_SERVER_ERROR, e.getMessage(), requestHandler);
        }
    }

    /**
     * Parse the query parameters using our defined BYTE_ENCODING.
     * <p>
     * <p>
     * Because we're expecting byte-encoded strings as query parameters, we
     * can't rely on SimpleHTTP's QueryParser which uses the wrong encoding for
     * the job and returns us unparsable byte data. We thus have to implement
     * our own little parsing method that uses BYTE_ENCODING to decode
     * parameters from the URI.
     * </p>
     * <p>
     * <p>
     * <b>Note:</b> array parameters are not supported. If a key is present
     * multiple times in the URI, the latest value prevails. We don't really
     * need to implement this functionality as this never happens in the
     * Tracker HTTP protocol.
     * </p>
     *
     * @param uri
     * @param hostAddress
     * @return The {@link AnnounceRequestMessage} representing the client's
     * announce request.
     */
    private HTTPAnnounceRequestMessage parseQuery(final String uri, final String hostAddress, String passkey)
            throws IOException, MessageValidationException {
        Map<String, BEValue> params = new HashMap<String, BEValue>();

        try {
//			String uri = request.getAddress().toString();
            for (String pair : uri.split("[?]")[1].split("&")) {
                String[] keyval = pair.split("[=]", 2);
                if (keyval.length == 1) {
                    this.recordParam(params, keyval[0], null);
                } else {
                    this.recordParam(params, keyval[0], keyval[1]);
                }
            }
        } catch (ArrayIndexOutOfBoundsException e) {
            params.clear();
        }

        // Make sure we have the peer IP, fallbacking on the request's source
        // address if the peer didn't provide it.
        if (params.get("ip") == null) {
            params.put("ip", new BEValue(
                    hostAddress,
                    Constants.BYTE_ENCODING));
        }

        return HTTPAnnounceRequestMessage.parse(new BEValue(params), passkey);
    }

    private void recordParam(Map<String, BEValue> params, String key, String value) {
        try {
            value = URLDecoder.decode(value, Constants.BYTE_ENCODING);

            for (String f : NUMERIC_REQUEST_FIELDS) {
                if (f.equals(key)) {
                    params.put(key, new BEValue(Long.valueOf(value)));
                    return;
                }
            }

            params.put(key, new BEValue(value, Constants.BYTE_ENCODING));
        } catch (UnsupportedEncodingException uee) {
            // Ignore, act like parameter was not there
        }
    }

    /**
     * Write a {@link HTTPTrackerErrorMessage} to the response with the given
     * HTTP status code.
     *
     * @param status The HTTP status code to return.
     * @param error  The error reported by the tracker.
     */
    private void serveError(Status status, HTTPTrackerErrorMessage error, RequestHandler requestHandler) throws IOException {
        requestHandler.serveResponse(status.getCode(), status.getDescription(), error.getData());
    }

    /**
     * Write an error message to the response with the given HTTP status code.
     *
     * @param status The HTTP status code to return.
     * @param error  The error message reported by the tracker.
     */
    private void serveError(Status status, String error, RequestHandler requestHandler) throws IOException {
        this.serveError(status, HTTPTrackerErrorMessage.craft(error), requestHandler);
    }

    /**
     * Write a tracker failure reason code to the response with the given HTTP
     * status code.
     *
     * @param status The HTTP status code to return.
     * @param reason The failure reason reported by the tracker.
     */
    private void serveError(Status status, ErrorMessage.FailureReason reason, RequestHandler requestHandler) throws IOException {
        this.serveError(status, reason.getMessage(), requestHandler);
    }

    public void setAcceptForeignTorrents(boolean acceptForeignTorrents) {
        myAcceptForeignTorrents = acceptForeignTorrents;
    }

    public interface RequestHandler {
        void serveResponse(int code, String description, ByteBuffer responseData);
    }
}
